package com.ls.fw.data.hbase.client.base.impl;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ls.fw.data.hbase.client.base.BaseHbaseDao;
import com.ls.fw.data.hbase.client.exception.HbaseException;

public class BaseHbaseDaoImpl implements BaseHbaseDao {

	protected Logger logger = LoggerFactory.getLogger(this.getClass());
	
	protected Configuration conf = null;
	
	protected HConnection connection = null;
	
	private boolean confMod =  false;    
	
	public BaseHbaseDaoImpl(String clientPort, String quorum, String master){
		this.changeConfig(clientPort, quorum, master);
	}
	
	@Override
	public void changeConfig(String clientPort, String quorum, String master){
		if(conf==null){
			conf = HBaseConfiguration.create();
		}
		conf.set("hbase.zookeeper.property.clientPort", clientPort);
		conf.set("hbase.zookeeper.quorum", quorum);
		conf.set("hbase.master", master); 
		this.confMod = true;
	}
	 

	@Override
	public void changeClientPort(String clientPort) {
		conf.set("hbase.zookeeper.property.clientPort", clientPort);
		this.confMod = true; 
	}
	
	
	@Override
	public void changeMaster(String master) {
		conf.set("hbase.master", master);
		this.confMod = true;
	}
	
	@Override
	public void changeQuorum(String quorum) {
		conf.set("hbase.zookeeper.quorum", quorum);
		this.confMod = true;
	}
	
	
	public synchronized void open() throws IOException{
		if(confMod || connection==null){
			this.close(connection);
			this.getNewConnection();
		}
	}
	
	
	@Override
	public void close() {
		this.close(connection);
		confMod = false;
	}
	
	public Configuration getConf() throws IOException{
		return conf;
	}
	
	public HConnection getNewConnection() throws IOException {
    	return connection = HConnectionManager.createConnection(this.getConf());
    }
    
    public HConnection getConnection() throws IOException {
    	if(connection==null){
     	   connection = getNewConnection();
         }   
    	return connection;
    }

    public Connection creatConnection(Configuration conf ) throws IOException {
    	 Connection connection = null;
    	 connection = ConnectionFactory.createConnection(conf);
    	return connection;
    }
    
    
    public HBaseAdmin creatHBaseAdmin() throws IOException {
    	HBaseAdmin admin = null;
    	admin = new HBaseAdmin(this.getConf());
    	return admin;
    }
    

    /**
     * 关闭连接
     */
    private void close(Connection connection ){
		if(connection!=null){
			try {
				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			connection = null;
		}
	}
   
	/**
	 * 关闭表
	 * @param htable
	 */
    private void close(HTableInterface  htable){
		if(htable!=null){
			try {
				htable.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			htable = null;
		}
		
	}
	
    private void close(HBaseAdmin  admin){
		if(admin!=null){
			try{
				admin.close();
			}catch (Exception e) {
				e.printStackTrace();
			}
			admin = null;
		}
	}
	
    
    public HTableInterface getTable(String tableName) throws IOException{
    	return this.getConnection().getTable(tableName);
    }
	
	/**
	 * 创建表操作 column familys add column <--->hbase columnName
	 * 
	 * @param tableName
	 * @param familys
	 * @throws Exception
	 */
	public void createTable(String tablename, String[] familys)
			throws HbaseException {
		HBaseAdmin admin = null;
		try {
			admin = this.creatHBaseAdmin();
			if (admin.tableExists(tablename)) {
				throw new HbaseException("表【"+tablename+"】已经存在！");
			} else {
				TableName tn = TableName.valueOf(tablename);
				HTableDescriptor tableDesc = new HTableDescriptor(tn);
				for (int i = 0; i < familys.length; i++) {
					tableDesc.addFamily(new HColumnDescriptor(familys[i]));
				}
				admin.createTable(tableDesc);
			}
		} catch (Exception e) {
			throw new HbaseException(e.getMessage(),e);
		}finally{
			this.close(admin);
		}
	}

	/**
	 * 删除表操作
	 * 
	 * @param tablename
	 * @throws ZooKeeperConnectionException
	 * @throws MasterNotRunningException
	 * @throws IOException
	 */
	public void deleteTable(String tableName) throws HbaseException{
		HBaseAdmin admin = null;
		try {
			admin = this.creatHBaseAdmin();
			if (admin.isTableEnabled(tableName)) {
				admin.disableTable(tableName);
			}
			admin.deleteTable(tableName);
		} catch (MasterNotRunningException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (ZooKeeperConnectionException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			this.close(admin);
		}

	}

	/**
	 * 删除一行记录
	 * 
	 * @param tablename
	 * @param rowkey
	 * @throws IOException
	 */
	public void deleteRow(String tablename, String rowkey) throws HbaseException {
		if (tablename == null || "".equals(tablename.trim())) {
			return;
		}
		HTableInterface table = null;
		try {
			table = this.getTable(tablename);
			this.deleteRow(table, rowkey);
		} catch (Exception e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			this.close(table);
		}
	}

	public void deleteRow(HTableInterface table, String rowkey)
			throws HbaseException {
		if (table == null) {
			return;
		}
		List<Delete> list = new ArrayList<Delete>();
		Delete d1 = new Delete(rowkey.getBytes());
		list.add(d1);
		try {
			table.delete(list);
		} catch (Exception e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
		}
	}

	/**
	 * 查找一行记录
	 * 
	 * @param tablename
	 * @param rowkey
	 */
	public Result selectRow(String tablename, String rowKey) throws HbaseException {
		if (tablename == null || "".equals(tablename.trim())) {
			return null;
		}
		HTableInterface htable = null;
		Result rs = null;
		try {
			htable = this.getTable(tablename);
			rs = selectRow(htable, rowKey);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			this.close(htable);
		}

		return rs;
	}

	/**
	 * 查找一行记录
	 * 
	 * @param table
	 * @param rowkey
	 */
	public Result selectRow(HTableInterface table, String rowKey)
			throws HbaseException {
		if (table == null) {
			return null;
		}
		Get g = new Get(rowKey.getBytes());
		Result rs = null;
		try {
			rs = table.get(g);
		} catch (Exception e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
		}
		return rs;
	}

	/**
	 * 查找一行记录
	 * 
	 * @param table
	 * @param rowkey
	 */
	public Result selectRow(HTableInterface table, Get get) throws HbaseException {
		if (table == null) {
			return null;
		}
		Result rs = null;
		try {
			rs = table.get(get);
		} catch (Exception e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
		}
		return rs;
	}

	/**
	 * 查询符合条件的行
	 * 
	 * @param tablename
	 * @param scan
	 * @return <rowkey,KeyValue[]>
	 * @throws IOException
	 */
	public NavigableMap<byte[], Cell[]> find(String tablename, Scan scan)
			throws HbaseException {
		if (tablename == null || "".equals(tablename.trim())) {
			return null;
		}
		NavigableMap<byte[], Cell[]> map = null;
		HTableInterface htable = null;
		try {
			htable = this.getTable(tablename); // instantiate HTable
			map = this.find(htable, scan);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			this.close(htable);
		}
		return map;
	}

	/**
	 * 查询符合条件的行
	 * 
	 * @param htable
	 * @param scan
	 * @return
	 * @throws IOException
	 */
	public NavigableMap<byte[], Cell[]> find(HTableInterface htable,
			Scan scan) throws HbaseException {
		if (htable == null) {
			return null;
		}
		ResultScanner rs = null;
		NavigableMap<byte[], Cell[]> map = null;
		try {
			rs = htable.getScanner(scan);
			map = new ConcurrentSkipListMap<byte[], Cell[]>();
			for (Result r = rs.next(); r != null; r = rs.next()) {
				Cell[] cells = r.rawCells();
				map.put(r.getRow(), cells);
			}
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			if (rs != null) {
				rs.close();
				rs = null;
			}
		}
		return map;
	}

	/**
	 * 查询符合条件的结果
	 * 
	 * @param tablename
	 * @param scan
	 * @return
	 * @throws IOException
	 */
	public List<Result> findResult(String tablename, Scan scan)
			throws HbaseException {
		if (tablename == null || "".equals(tablename.trim())) {
			return null;
		}
		List<Result> list = null;
		HTableInterface htable = null;
		try {
			htable = new HTable(this.getConf(), tablename); // instantiate HTable
			list = this.findResult(htable, scan);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			this.close(htable);
		}
		return list;
	}

	/**
	 * 查询符合条件的结果
	 * 
	 * @param htable
	 * @param scan
	 * @return
	 * @throws IOException
	 */
	public List<Result> findResult(HTableInterface htable, Scan scan)
			throws HbaseException {
		if (htable == null) {
			return null;
		}
		ResultScanner rs = null;
		List<Result> list = null;
		try {
			long time = 0;
			long startTime = System.currentTimeMillis(); // 获取开始时间
			System.out.println("开始查询。。。");
			rs = htable.getScanner(scan);
			list = new ArrayList<Result>();
			int size = 0;
			for (Result r = rs.next(); r != null; r = rs.next()) {
				list.add(r);
				size++;
			}
			long endTime = System.currentTimeMillis(); // 获取结束时间
			time = endTime - startTime;
			System.out.println("查询 " + size + " 条数据，" + "程序运行时间： " + (time)
					+ "ms");
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			if (rs != null) {
				rs.close();
				rs = null;
			}
		}
		return list;
	}

	/**
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @param value
	 * @throws Exception
	 */
	public void putData(String tableName, String rowKey, String family,
			String qualifier, String value) throws HbaseException {
		this.putData(tableName, rowKey, family, qualifier, value, null);
	}

	public void putData(HTableInterface ht, String rowKey, String family,
			String qualifier, String value) throws HbaseException {
		this.putData(ht, rowKey, family, qualifier, value, null);
	}

	public void putData(String tableName, String rowKey, String family,
			String qualifier, String value, Long version) throws HbaseException {
		HTableInterface ht = null; 
		try {
			ht = this.getTable(tableName);
			ht.setAutoFlush(false,true); // 数据入库之前先设置此项为false
			this.putData(ht, rowKey, family, qualifier, value, version);
			ht.flushCommits();// 入库完成后，手动刷入数据
		} catch (RetriesExhaustedWithDetailsException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (InterruptedIOException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally{
			this.close(ht);
		}
	}

	/**
	 * 说明： 一个 {row, column, version} 元组是HBase中的一个单元(cell).
	 * 但是有可能会有很多的单元的行和列是相同的，可以使用版本来区分不同的单元
	 * 
	 * @param ht
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @param value
	 * @param version
	 * @throws Exception
	 */
	public void putData(HTableInterface ht, String rowKey, String family,
			String qualifier, String value, Long version) throws HbaseException {
		Put put = new Put(Bytes.toBytes(rowKey));
		// 在入库过程中，
		put.setDurability(Durability.SYNC_WAL);
		// 关于这一项如果不希望大量数据在存储过程中丢失，建议设置为true,如果仅是在测试演练阶段，为了节省入库时间建议设置为false
		if (version == null) {
			// 用当前时间作为版本
			put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes
					.toBytes(value));
		} else {
			// 用version作为版本
			put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), version
					.longValue(), Bytes.toBytes(value));
		}
		try {
			ht.put(put);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		}
	}

	public void putData(HTableInterface ht, KeyValue keyValue) throws HbaseException {
		Put put = new Put(keyValue.getKey());
		try {
			put.add(keyValue);
			ht.put(put);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		}
	}

	public void putData(String tableName, KeyValue keyValue) throws HbaseException {
		HTable ht = null;
		try {
			ht = new HTable(this.getConf(), tableName);
			this.putData(ht, keyValue);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		}finally{
			this.close(ht);
		}
	}

	/**
	 * 添加列
	 * 
	 * @param tableName
	 * @param column
	 * @throws IOException
	 */
	public void addColumn(String tableName, HColumnDescriptor column)
			throws HbaseException {
		HBaseAdmin admin = null;
		try {
			admin = this.creatHBaseAdmin();
			if (admin.isTableEnabled(tableName)) {
				admin.disableTable(tableName);
			}
			admin.addColumn(tableName, column); // adding new ColumnFamily
			admin.enableTable(tableName);
		} catch (ZooKeeperConnectionException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (MasterNotRunningException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			if (admin != null) {
				try {
					if (!admin.isTableEnabled(tableName)) {
						admin.enableTable(tableName);
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			this.close(admin);
		}

	}

	/**
	 * 修改列
	 * 
	 * @param tableName
	 * @param column
	 * @throws IOException
	 */
	public void modifyColumn(String tableName, HColumnDescriptor column)
			throws HbaseException {
		HBaseAdmin admin = null;
		try {
			admin = this.creatHBaseAdmin();
			if (admin.isTableEnabled(tableName)) {
				admin.disableTable(tableName);
			}
			admin.modifyColumn(tableName, column); // adding new ColumnFamily
			admin.enableTable(tableName);
		} catch (ZooKeeperConnectionException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (MasterNotRunningException e) {
			throw new HbaseException(e.getMessage(),e);
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			if (admin != null) {
				try {
					if (!admin.isTableEnabled(tableName)) {
						admin.enableTable(tableName);
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			this.close(admin);
		}

	}

	@Override
	public int getRowCount(String tablename) throws HbaseException {
		return this.getRowCount(tablename, new Scan());
	}

	@Override
	public int getRowCount(String tableName, Scan scan) throws HbaseException {
		if (tableName == null || "".equals(tableName.trim())) {
			return 0;
		}
		ResultScanner rs = null;
		HTableInterface htable = null;
		int count = 0;
		try {
			htable = new HTable(this.getConf(), tableName); // instantiate HTable
			rs = htable.getScanner(scan);
			for (Result r = rs.next(); r != null; r = rs.next()) {
				count++;
			}
		} catch (IOException e) {
			throw new HbaseException(e.getMessage(),e);
		} finally {
			if (rs != null) {
				rs.close();
				rs = null;
			}
			this.close(htable);
		}
		return count;
	}

	
	public String getCellData(String tableName, String rowName, String columnFamilyName, String columnName) {
        String cellData = "";
        try (HTable hTable = new HTable(this.getConf(), tableName)) {

            // Instantiating Get class
            Get g = new Get(Bytes.toBytes(rowName));

            // Reading the data
            Result result = hTable.get(g);

            byte[] value = result.getValue(Bytes.toBytes(columnFamilyName), Bytes.toBytes(columnName));
            cellData = Bytes.toString(value);
        } catch (IOException e) {
        	throw new HbaseException(e.getMessage(),e);
        }
        return cellData;
    }

	
	public void updateCell(String tableName, String rowName, String columnFamilyName, String columnName, String value) {
        try (HTable hTable = new HTable(this.getConf(), tableName)) {
            Put p = new Put(Bytes.toBytes(rowName));
            p.add(Bytes.toBytes(columnFamilyName), Bytes.toBytes(columnName), Bytes.toBytes(value));
            hTable.put(p);
        } catch (IOException e) {
        	throw new HbaseException(e.getMessage(),e);
        }
    }
	
	public List<Map<String,Object>> query(String tableName,String... id){
		Map<String,Object> map = new HashMap<String,Object>();
		
		
		
		return null;
	}
}
